//
// Created by amon.huang on 2021/7/26.
//

#include <unistd.h>
#include "PushPacketQueue.h"
#include "PushConstDefine.h"

void PushPacketQueue::clear() {
    pthread_mutex_lock(&pushPacketMutex);
    if (pPushPacketQueue == nullptr) {
        return;
    }
    while (!pPushPacketQueue->empty()) {
        RTMPPacket *pRTMPPacket = pPushPacketQueue->front();
        pPushPacketQueue->pop();
        RTMPPacket_Free(pRTMPPacket);
        free(pRTMPPacket);
    }
    pthread_mutex_unlock(&pushPacketMutex);
}

RTMPPacket *PushPacketQueue::pop() {
    RTMPPacket *pPacket = nullptr;
    pthread_mutex_lock(&pushPacketMutex);
    if (pPushPacketQueue->empty()) {
        pthread_cond_wait(&pushPacketCond, &pushPacketMutex);
    } else {
        pPacket = pPushPacketQueue->front();
        pPushPacketQueue->pop();
    }
    pthread_mutex_unlock(&pushPacketMutex);
    return pPacket;
}

PushPacketQueue::PushPacketQueue() {
    pPushPacketQueue = new std::queue<RTMPPacket *>();
    pthread_mutex_init(&pushPacketMutex, nullptr);
    pthread_cond_init(&pushPacketCond, nullptr);
}

PushPacketQueue::~PushPacketQueue() {
    if (pPushPacketQueue) {
        clear();
        delete pPushPacketQueue;
        pPushPacketQueue = nullptr;
        pthread_mutex_destroy(&pushPacketMutex);
        pthread_cond_destroy(&pushPacketCond);
    }
}

void PushPacketQueue::push(RTMPPacket *pPacket) {
    pthread_mutex_lock(&pushPacketMutex);
    pPushPacketQueue->push(pPacket);
    pthread_cond_signal(&pushPacketCond);
    pthread_mutex_unlock(&pushPacketMutex);
}

void PushPacketQueue::notify() {
    pthread_mutex_lock(&pushPacketMutex);
    pthread_cond_signal(&pushPacketCond);
    pthread_mutex_unlock(&pushPacketMutex);
}
